package org.myfram.flink.flinkonjar.common.connector.kafka.sink;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkFixedPartitioner;
import org.myfram.flink.flinkonjar.domain.entity.KafkaTargetUserEvent;

import java.util.Optional;
import java.util.Properties;

public class KafkaSinkBuilder {

    public static SinkFunction<String> build(String topic, Properties props) {
        return new FlinkKafkaProducer<String>(
                topic,
                new SimpleStringSchema(),
                props, Optional.of(new FlinkFixedPartitioner<>())
        );
    }

    public static String serialize(KafkaTargetUserEvent event) {
        try {
            return new ObjectMapper().writeValueAsString(event);
        } catch (JsonProcessingException e) {
            throw new RuntimeException(e);
        }
    }

}
